package com.cloudbroker.bcs.common.mycat.route.function;

import com.cloudbroker.bcs.common.mycat.config.model.rule.RuleAlgorithm;
import com.cloudbroker.bcs.common.mycat.util.exception.MurmurHashException;
import com.cloudbroker.bcs.common.util.ValueUtil;
import com.google.common.hash.HashFunction;
import com.google.common.hash.Hashing;

import java.io.*;
import java.util.*;

/**
 * Created by xuyn15828 on 2016/6/7.
 */
public class PartitionByMurmurHash extends AbstractPartitionAlgorithm implements RuleAlgorithm {

    private int seed = 0;
    private int count;
    private int virtualBucketTimes = 160;
    private Map<Integer, Integer> weightMap = new HashMap();
    private HashFunction hash;
    private SortedMap<Integer, Integer> bucketMap;

    public void init() {
        try {
            this.bucketMap = new TreeMap();
            generateBucketMap();
        } catch (Exception e) {
            throw new MurmurHashException(e);
        }
    }

    private void generateBucketMap() {
        this.hash = Hashing.murmur3_32(this.seed);
        for (int i = 0; i < this.count; i++) {
            StringBuilder hashName = new StringBuilder("SHARD-").append(i);
            int n = 0;
            for (int shard = this.virtualBucketTimes * getWeight(i); n < shard; n++) {
                this.bucketMap.put(Integer.valueOf(this.hash.hashUnencodedChars(hashName.append("-NODE-").append(n)).asInt()), Integer.valueOf(i));
            }
        }
        this.weightMap = null;
    }

    private int getWeight(int bucket) {
        return ValueUtil.getInt(this.weightMap.get(Integer.valueOf(bucket)) , 1);
    }

    public void setCount(int count) {
        this.count = count;
    }

    public Integer calculate(String columnValue) {
        SortedMap tail = this.bucketMap.tailMap(Integer.valueOf(this.hash.hashUnencodedChars(columnValue).asInt()));
        if (tail.isEmpty()) {
            return this.bucketMap.get(this.bucketMap.firstKey());
        }
        return (Integer) tail.get(tail.firstKey());
    }

    private static void getTargetNode(int dnCount, String[] colValue) throws IOException {
        PartitionByMurmurHash hash = new PartitionByMurmurHash();
        hash.count = dnCount;
        hash.init();
        for (int i = 1; i < colValue.length; i++) {
            int h = hash.calculate(colValue[i]).intValue();
            System.out.println("value:" + colValue[i] + " datanode:" + h);
        }
    }

    public static Integer caluateMurmurHash(int dnCount , String colValue) {
        PartitionByMurmurHash hash = new PartitionByMurmurHash();
        hash.count = dnCount;
        hash.init();
        return hash.calculate(colValue).intValue();
    }

    public static void main(String[] args) throws IOException {
        /*Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()) {
            String tenantId = scanner.next();
            System.out.println(tenantId + "=" + caluateMurmurHash(8 , tenantId));
        }*/
        System.out.println(caluateMurmurHash(4 , "70004"));
    }
}
